package io.debezium.connector.postgresql;

import io.debezium.config.Configuration;
import io.debezium.connector.common.RelationalBaseSourceConnector;
import io.debezium.connector.postgresql.connection.GaussDBConnection;
import io.debezium.relational.RelationalDatabaseConnectorConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigValue;
import org.apache.kafka.connect.connector.Task;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLException;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class PostgresConnector extends RelationalBaseSourceConnector {
    private static final Logger LOGGER = LoggerFactory.getLogger(PostgresConnector.class);
    private Map<String, String> props;

    public PostgresConnector() {}

    public String version() {
        return Module.version();
    }

    public Class<? extends Task> taskClass() {
        return PostgresConnectorTask.class;
    }

    public void start(Map<String, String> props) {
        this.props = props;
    }

    public List<Map<String, String>> taskConfigs(int maxTasks) {
        return this.props == null
                ? Collections.emptyList()
                : Collections.singletonList(new HashMap(this.props));
    }

    public void stop() {
        this.props = null;
    }

    public ConfigDef config() {
        return PostgresConnectorConfig.configDef();
    }

    protected void validateConnection(Map<String, ConfigValue> configValues, Configuration config) {
        ConfigValue databaseValue =
                configValues.get(RelationalDatabaseConnectorConfig.DATABASE_NAME.name());
        ConfigValue slotNameValue = configValues.get(PostgresConnectorConfig.SLOT_NAME.name());
        ConfigValue pluginNameValue = configValues.get(PostgresConnectorConfig.PLUGIN_NAME.name());
        if (databaseValue.errorMessages().isEmpty()
                && slotNameValue.errorMessages().isEmpty()
                && pluginNameValue.errorMessages().isEmpty()) {
            PostgresConnectorConfig postgresConfig = new PostgresConnectorConfig(config);
            ConfigValue hostnameValue =
                    configValues.get(RelationalDatabaseConnectorConfig.HOSTNAME.name());
            GaussDBConnection connection =
                    new GaussDBConnection(
                            postgresConfig.getJdbcConfig(), "Debezium Validate Connection");

            try {
                try {
                    connection.connection(false);
                    connection.execute(new String[] {"SELECT version()"});
                    LOGGER.info(
                            "Successfully tested connection for {} with user '{}'",
                            connection.connectionString(),
                            connection.username());
                    String walLevel =
                            connection.queryAndMap(
                                    "SHOW wal_level",
                                    connection.singleResultMapper(
                                            (rs) -> rs.getString("wal_level"),
                                            "Could not fetch wal_level"));
                    String errorMessage;
                    if (!"logical".equals(walLevel)) {
                        errorMessage =
                                "Postgres server wal_level property must be \"logical\" but is: "
                                        + walLevel;
                        LOGGER.error(errorMessage);
                        hostnameValue.addErrorMessage(errorMessage);
                    }

                    if (!(Boolean)
                            connection.queryAndMap(
                                    "SELECT r.rolcanlogin AS rolcanlogin, r.rolreplication AS rolreplication, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rds_superuser') AS BOOL) IS TRUE AS aws_superuser, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsadmin') AS BOOL) IS TRUE AS aws_admin, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsrepladmin') AS BOOL) IS TRUE AS aws_repladmin, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rds_replication') AS BOOL) IS TRUE AS aws_replication FROM pg_roles r WHERE r.rolname = current_user",
                                    connection.singleResultMapper(
                                            (rs) -> {
                                                return rs.getBoolean("rolcanlogin")
                                                        && (rs.getBoolean("rolreplication")
                                                                || rs.getBoolean("aws_superuser")
                                                                || rs.getBoolean("aws_admin")
                                                                || rs.getBoolean("aws_repladmin")
                                                                || rs.getBoolean(
                                                                        "aws_replication"));
                                            },
                                            "Could not fetch roles"))) {
                        errorMessage =
                                "Postgres roles LOGIN and REPLICATION are not assigned to user: "
                                        + connection.username();
                        LOGGER.error(errorMessage);
                    }
                } catch (SQLException var12) {
                    SQLException e = var12;
                    LOGGER.error(
                            "Failed testing connection for {} with user '{}'",
                            new Object[] {connection.connectionString(), connection.username(), e});
                    hostnameValue.addErrorMessage(
                            "Error while validating connector config: " + e.getMessage());
                }
            } catch (Throwable var13) {
                try {
                    connection.close();
                } catch (Throwable var11) {
                    var13.addSuppressed(var11);
                }

                throw var13;
            }

            connection.close();
        }
    }

    protected Map<String, ConfigValue> validateAllFields(Configuration config) {
        return config.validate(PostgresConnectorConfig.ALL_FIELDS);
    }
}
